package yuan.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import yuan.Company;
import yuan.producer.intepceptor.ProducerInterceptorPrefix;
import yuan.producer.partition.DemoPartitioner;

import yuan.serializer.CompanySerializer;

import java.util.Properties;

import static yuan.consumer.interceptor.ConsumerInterceptorTTL.EXPLAIN_INTERVAL;

public class KafkaProducerAnalysis {
    public static final String brokerList = "192.168.190.128:9092";
    public static final String topic = "topic-demo";

    public static Properties init(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer.client.id.demo");
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorPrefix.class.getName());
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DemoPartitioner.class.getName());
        return properties;

    }

    public static void main(String[] args) {

        Properties properties = init();


        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        try {
            /*//同步
            RecordMetadata recordMetadata = producer.send(record).get();*/
            //异步 回调
            for (int i = 0; i < 10; i++) {
//                Company company = new Company("name" + String.valueOf(1), "address" + String.valueOf(1));
                ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, 0, System.currentTimeMillis() - EXPLAIN_INTERVAL - 1, null, "expire-message" + String.valueOf(i));

                producer.send(record, (recordMeta, exception) -> {
                    System.out.println("hasOffset:"+recordMeta.hasOffset());
                    System.out.println("offset"+recordMeta.offset());
                    System.out.println("partition:"+recordMeta.partition());
                });
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
        producer.close();
    }
}
